package cn.dmp.charts.arealDistribution

import cn.dmp.utils.ParseUtils
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.rdd.RDD

/**
  * create by lfq on 171106
  */

/**
  * 思路一。每个指标按省市聚合。然后几个rdd  join
  *
  * 思路二。
  */
object ArealDistribution171106V2 {
  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf().setAppName("ArealDistribution171106V2").setMaster("local[4]")
    conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
      .set("spark.sql.parquet.compression.codec", "snappy")

    val sc: SparkContext = new SparkContext(conf)

    val rawRDD: RDD[String] = sc.textFile(args(0))

    val splitsRDD: RDD[Array[String]] = rawRDD.map(_.split(",",-1))

    val suitedLengthRDD = splitsRDD.filter(_.length == 85)

    val baseRDD = suitedLengthRDD.map(splits => {

      val provincename: String = splits(24)
      val cityname: String = splits(25)
      val requestmode: Int = ParseUtils.parseInt(splits(8))
      val processnode: Int = ParseUtils.parseInt(splits(35))
      val iseffective: Int = ParseUtils.parseInt(splits(30))
      val isbilling: Int = ParseUtils.parseInt(splits(31))
      val isbid: Int = ParseUtils.parseInt(splits(39))
      val iswin: Int = ParseUtils.parseInt(splits(42))
      val adorderid: Int = ParseUtils.parseInt(splits(2))
      val winprice: Double = ParseUtils.parseDouble(splits(41))
      val adpayment: Double = ParseUtils.parseDouble(splits(75))

      (provincename, cityname, requestmode, processnode, iseffective, isbilling, isbid, iswin, adorderid, winprice, adpayment)

    })

    baseRDD.cache()

    //原始请求数primaryRequest
    val primaryRequestRDD = baseRDD.map(t => {(t._1, t._2, t._3,t._4)})
      .filter(t => {t._3 == 1 && t._4 >= 1})
      .map(t=>{((t._1,t._2),1)}).reduceByKey(_+_)


//    primaryRequestRDD.foreach(t=>{
//      println(t._1._1+"-"+t._1._2+":"+t._2)
//    })

//      .groupBy(t=>{(t._1,t._2)})
    //有效请求书effectiveRequest
    val effectiveRequestRDD = baseRDD.map(t => {(t._1, t._2, t._3,t._4)})
      .filter(t => {t._3 == 1 && t._4 >= 2})
      .map(t=>{((t._1,t._2),1)}).reduceByKey(_+_)
    //广告请求书AdRequest
    val AdRequestRDD = baseRDD.map(t => {(t._1, t._2, t._3,t._4)})
      .filter(t => {t._3 == 1 && t._4 >= 3})
      .map(t=>{((t._1,t._2),1)}).reduceByKey(_+_)


    //参与竞价次数biddingTimes
    val biddingTimesRDD = baseRDD.map(t => {(t._1, t._2, t._5,t._6,t._7,t._9)})
      .filter(t => {t._3==1 && t._4==1 && t._5==1 && t._6 !=0 })
      .map(t=>{((t._1,t._2),1)}).reduceByKey(_+_)
    //成功竞价次数succesedBiddingTime
    val succesedBiddingTimeRDD = baseRDD.map(t => {(t._1, t._2, t._5,t._6,t._8)})
      .filter(t => {t._3==1 && t._4==1 && t._5==1 })
      .map(t=>{((t._1,t._2),1)}).reduceByKey(_+_)
    //展示量showTimes
    val showTimesRDD = baseRDD.map(t => {(t._1, t._2, t._3,t._5)})
      .filter(t => {t._3 == 2 && t._4 == 1})
      .map(t=>{((t._1,t._2),1)}).reduceByKey(_+_)
    //点击量clickTimes
    val clickTimesRDD = baseRDD.map(t => {(t._1, t._2, t._3,t._5)})
      .filter(t => {t._3 == 3 && t._4 == 1})
      .map(t=>{((t._1,t._2),1)}).reduceByKey(_+_)
    //广告消费AdConsume
    val AdConsumeRDD = baseRDD.map(t => {(t._1, t._2, t._5,t._6,t._8,t._10)})
      .filter(t => {t._3==1 && t._4==1 && t._5==1 })
      .map(t=>{((t._1,t._2),t._6)}).reduceByKey(_+_)
    //广告成本AdCost
    val AdCostRDD = baseRDD.map(t => {(t._1, t._2, t._5,t._6,t._8,t._11)})
      .filter(t => {t._3==1 && t._4==1 && t._5==1 })
      .map(t=>{((t._1,t._2),t._6)}).reduceByKey(_+_)


    //9个rdd都有相同的分组，可以join

  //方法一   join
    val join: RDD[((String, String), ((((((((Int, Int), Int), Int), Int), Int), Int), Double), Double))] = primaryRequestRDD
      .join(effectiveRequestRDD)
      .join(AdRequestRDD)
      .join(biddingTimesRDD)
      .join(succesedBiddingTimeRDD)
      .join(showTimesRDD)
      .join(clickTimesRDD)
      .join(AdConsumeRDD)
      .join(AdCostRDD)

    val cuple: RDD[(String, String, Int, Int, Int, Int, Int, Int, Int, Double, Double)] = join.map(t => {
      (
        t._1._1,
        t._1._2,
        t._2._1._1._1._1._1._1._1._1,
        t._2._1._1._1._1._1._1._1._2,
        t._2._1._1._1._1._1._1._2,
        t._2._1._1._1._1._1._2,
        t._2._1._1._1._1._2,
        t._2._1._1._1._2,
        t._2._1._1._2,
        t._2._1._2,
        t._2._2
      )
    })





  }
}
